package edu.gmu.router;

import edu.gmu.messages.Client;
import edu.gmu.messages.Connect;
import edu.gmu.messages.Disconnect;
import edu.gmu.messages.Subscribe;
import edu.gmu.messages.Unsubscribe;
import edu.gmu.middleware.Header;
import edu.gmu.middleware.Message;
import edu.gmu.middleware.SinkFactory;
import edu.gmu.middleware.Sink;
import edu.gmu.middleware.SinkDataEvent;
import edu.gmu.middleware.SinkDisconnectedEvent;
import edu.gmu.middleware.SinkListener;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;

/**
 * This class is the TCP server process for a MOM Publish/Subscribe system.
 * @author James Pope
 */
public class Router implements RouterDispatcherListener
{
    /**
     * The number of byte buffer starting in the pool (may grow?)
     */
    public static final int DEFAULT_INPOOL_SIZE = 10;
    
    /**
     * The size of the byte buffers in the pool - all the same size
     */
    public static final int DEFAULT_INBUFFER_SIZE = 1048576;
    
    
    private ServerSocket serverSocket = null;
    private int port = 0;
    private volatile boolean stopped = false;
    
    //------------------------------------------------------------------------//
    // These two data structures should be in tandem (no client in one
    // and not the other) - we need to add some synchroniztion so
    // that we are not updating while another thread is reading
    //------------------------------------------------------------------------//
    /**
     * List of clients
     */
    private List clients = new LinkedList();
    
    /**
     * Map of subscriptions - each has list of subscribed clients
     */
    private LinkedHashMap<Integer,List<ClientEntry>> subscriptions =
            new LinkedHashMap<Integer,List<ClientEntry>>();
    //------------------------------------------------------------------------//
    
    
    private RouterDispatcher dispatcher = null;
    
    /**
     * Creates a new Router with specified params.
     */
    public Router( )
    {
        this.dispatcher = new RouterDispatcher( DEFAULT_INPOOL_SIZE,
                                                DEFAULT_INBUFFER_SIZE );
        
        this.dispatcher.register(RouterMessage.CONNECT,     Connect.class);
        this.dispatcher.register(RouterMessage.DISCONNECT,  Disconnect.class);
        this.dispatcher.register(RouterMessage.SUBSCRIBE,   Subscribe.class);
        this.dispatcher.register(RouterMessage.UNSUBSCRIBE, Unsubscribe.class);
        
        this.dispatcher.addRouterDispatchListener(this);
        this.dispatcher.start();
    }
    
    /**
     * Starts the server listening on specified port - blocks caller's thread
     * until finished.
     * @param port
     * @throws java.io.IOException
     */
    public void listen( int port ) throws IOException
    {
        this.port = port;
        this.serverSocket = new ServerSocket(this.port);
        
        this.stopped = false;
        while( stopped == false && this.dispatcher.isStopped() == false )
        {
            // Blocks until client connects
            Socket clientSocket = this.serverSocket.accept();
            this.handleClient(clientSocket);
        }
        System.out.println("=========================================");
        System.out.println("= Listener is exiting - bye!");
        System.out.println("=========================================");
    }
    
    /**
     * Handles a new client connection.
     * @param clientSocket
     */
    public void handleClient( Socket clientSocket )
    {
        //--------------------------------------------------------------------//
        // Be quick here so that server can go back to listening
        //--------------------------------------------------------------------//
        try
        {
            final Sink sink = SinkFactory.makeTCPSink(clientSocket, DEFAULT_INBUFFER_SIZE);
            
            sink.addSinkListener( new SinkListener()  
                {
                    public void dataReceived( SinkDataEvent event )
                    {
                        
                    }
                    
                    public void sinkDisconnected( SinkDisconnectedEvent event )
                    {
                        // We know which client - therefore remove and close

                        // To be quite honest - not sure what "this" means here?
                        /*
                         * Get periodic IAOOBE when calling removeExceptionListener
                         * Race condition - but not sure where?
                         */
                        sink.removeSinkListener(this);
                        sink.removeSinkListener(dispatcher);
                        
                        System.out.println("Handling sinkDisconnected in TCPRouter.");
                        event.getException().printStackTrace();
                        
                        removeEntry(sink);
                        sink.disconnect();
                        //System.out.println("   clients.size()="+clients.size());
                    }
                }
            );

            sink.addSinkListener( dispatcher );
            
            //System.out.println("Router: adding client.");
            addEntry(sink);
            sink.connect();
        }
        catch(IOException ioe)
        {
            ioe.printStackTrace();
        }
        //--------------------------------------------------------------------//
    }
    
    /**
     * This method is called when a message is received.
     * @param event
     */
    public void forwardMessageEvent(ForwardMessageEvent event)
    {
        //--------------------------------------------------------------------//
        // Executed on Dispacther's Thread
        //--------------------------------------------------------------------//
        Header header       = event.getHeader();
        ByteBuffer inBuffer = event.getInBuffer();
        Sink sink           = event.getSink();
        
        //System.out.println("Received forward " + header.toString());
        
        // Interesting - opposite of drain - we prepare for writing
        // Usually it is prepare for reading after writing
        inBuffer.position( inBuffer.limit() );
        
        int topic = header.getTopic();
        List<ClientEntry> subscribedClients = this.subscriptions.get(topic);
        if( subscribedClients == null )
        {
            System.out.println( "No subscribers for "+topic );
            return;
        }
        
        for (int i = 0; i < subscribedClients.size(); i++)
        {
            ClientEntry clientEntry = subscribedClients.get(i);
            
            Client client       = clientEntry.getClient();
            Sink subscribedSink = clientEntry.getSink();
            
            //System.out.println( "Forwarding " + topic + " to client.id=" +
            //        client.id + " client.name="+client.name );
            
            // Do not send to source of message even is registered
            // i.e. source is subscribed for message it published.
            if( subscribedSink == sink )
            {
                //System.out.println( "Skipping client.id=" +
                //    client.id + " client.name="+client.name +
                //    " published topic " + topic );
                continue;
            }
            
            // Attempt to make sure that the clientSink is connected
            if( subscribedSink.isConnected() == true )
            {
                try
                {
                    //System.out.println( "Writing " + topic + " to client.id=" +
                    //client.id + " client.name="+client.name + " sink="+subscribedSink );
                    subscribedSink.write(inBuffer);
                    // Prepare to write again
                    inBuffer.position( inBuffer.limit() );
                }
                catch( IOException ioe )
                {
                    ioe.printStackTrace();
                }
            }
            else
            {
                /*
                 * SINCE WE HAVE DETECTED THIS - WHY WASN'T THIS DETECTED
                 */
                System.out.println( "Subscribed client not connected "+client );
            }
            
        }
        //--------------------------------------------------------------------//
    }
    
    /**
     * This method is called when a message is received.
     * @param event
     */
    public void routerMessageEvent(RouterMessageEvent event)
    {
        //--------------------------------------------------------------------//
        // Executed on Dispacther's Thread
        //--------------------------------------------------------------------//
        // Now figure out who is registered for this message
        Header header   = event.getHeader();
        Message message = event.getMessage();
        Sink sink       = event.getSink();
        int messageId = header.getTopic();
        //System.out.println( "RouterMessageEvent: messageId="+messageId );
        if( messageId == RouterMessage.CONNECT )
        {
            //System.out.println( "Connecting="+((Connect)message).client.name );
            handleConnect( (Connect)message, sink );
        }
        else if( messageId == RouterMessage.DISCONNECT )
        {
            handleDisconnect( (Disconnect)message );
        }
        else if( messageId == RouterMessage.SUBSCRIBE )
        {
            //System.out.println( "Subscribing="+((Subscribe)message).messageId );
            handleSubscribe( (Subscribe)message, sink );
        }
        else if( messageId == RouterMessage.UNSUBSCRIBE )
        {
            handleUnsubscribe( (Unsubscribe)message, sink );
        }
        //--------------------------------------------------------------------//
    }
    
    private int count = 1;
    
    /**
     * Adds the clientEntry for the given sink
     * @param sink
     * @return clientEntry or null if entry already exists
     */
    public ClientEntry addEntry( Sink sink )
    {
        if( this.findEntry( sink ) != null )
        {
            System.out.println( "Sink already exists - not adding" );
            return null;
        }
        
        Client client = new Client( );
        client.id   = count;
        
        client.name = "Client #"+count;
        count++;
        ClientEntry entry = new ClientEntry(client, sink);
        
        this.clients.add(entry);
        return entry;
    }
    
    /**
     * Finds the clientEntry for the given sink
     * @param sink
     * @return clientEntry or null if not found
     */
    public ClientEntry findEntry( Sink sink )
    {
        ClientEntry entry = null;
        for (int i = 0; i < clients.size(); i++)
        {
            ClientEntry clientEntry = (ClientEntry)clients.get(i);
            if( clientEntry.getSink() == sink )
            {
                entry = clientEntry;
                break;
            }
        }
        return entry;
    }
    
    /**
     * Removes the clientEntry for the given sink
     * @param sink
     * @return true if removed, false if not found
     */
    public void removeEntry( Sink sink )
    {
        /*
         * We have clients in two places - client list and the subscriptions
         */
        for (int i = 0; i < clients.size(); i++)
        {
            ClientEntry clientEntry = (ClientEntry)clients.get(i);
            if( clientEntry.getSink() == sink )
            {
                clients.remove( i );
                //break; // should only be one - but just to be sure
            }
        }
        Collection<List<ClientEntry>> values = this.subscriptions.values();
        Iterator<List<ClientEntry>> iter = values.iterator();
        while( iter.hasNext() )
        {
            List<ClientEntry> subscribedClients = iter.next();
            for (int j = 0; j < subscribedClients.size(); j++)
            {
                ClientEntry clientEntry = (ClientEntry)subscribedClients.get(j);
                if( clientEntry.getSink() == sink )
                {
                    subscribedClients.remove( j );
                    //break; // should only be one - but just to be sure
                }
            }
        }
    }
    
    
    
    /**
     * Handles Connect Message
     * @param message
     */
    public void handleConnect( Connect message, Sink sink )
    {
        Client client = message.client;
        //int clientId  = client.id;
        
        // Make sure we do not already have a connection for this Client
        // If so remove/close old one and assume it was orphaned
        //this.clients
        ClientEntry existingEntry = findEntry(sink);
        if( existingEntry != null && existingEntry.getClient().id == -1 )
        {
            Client clientEntry = existingEntry.getClient();
            clientEntry.id   = client.id;
            clientEntry.name = client.name;
            
            System.out.println( "Connected client.id=" +
                    client.id + " client.name="+client.name );
        }
        else
        {
            System.out.println( "Ignoring connect from client.id=" +
                    client.id + " client.name="+client.name );
        }
    }
    
    /**
     * Handles Disconnect Message
     * @param message
     */
    public void handleDisconnect( Disconnect message )
    {
        Client client = message.client;
        int clientId  = client.id;
        
        // Make sure we do not already have a connection for this Client
        // If so remove/close old one and assume it was orphaned
        //this.clients
        ClientEntry existingEntry = (ClientEntry)this.clients.get( client.id );
        if( existingEntry != null )
        {
            this.clients.remove(existingEntry.getClient().id);
        }
        else
        {
            System.out.println( "Ignoring connect from client.id=" +
                    client.id + " client.name="+client.name );
        }
    }
    
    /**
     * Handles Subscribe Message
     * @param message
     */
    public void handleSubscribe( Subscribe message, Sink sink )
    {
        ClientEntry entry = this.findEntry(sink);
        
        if( entry == null )
        {
            System.out.println("Unknown subscribing client " + message.client.name);
            return;
        }
        
        Client subscribedClient = entry.getClient();
        
        // Find subscriptions for topic (if necessary create list) and add
        int[] topics = message.topics;
        for (int i = 0; i < topics.length; i++)
        {
            int topic = topics[i];
            List<ClientEntry> subscribedClients = this.subscriptions.get(topic);
            if( subscribedClients == null )
            {
                subscribedClients = new LinkedList<ClientEntry>();
                this.subscriptions.put( topic, subscribedClients);
            }
            
            if( !subscribedClients.contains(subscribedClient) )
            {
                System.out.println("Client subscriber for " + topic);
                subscribedClients.add( entry );
            }
            else
            {
                System.out.println("Client already subscribed=" +
                        subscribedClient.name +
                        " topic="+topic);
            }
        }
    }
    
    /**
     * Handles Unsubscribe Message
     * @param message
     * @param sink
     */
    public void handleUnsubscribe( Unsubscribe message, Sink sink )
    {
        ClientEntry entry = this.findEntry(sink);
        
        if( entry == null )
        {
            System.out.println("Unknown subscribing client " + message.client.name);
            return;
        }
        
        Client subscribedClient = entry.getClient();
        
        // Find subscriptions for topic (if necessary create list) and add
        int[] topics = message.topics;
        for (int i = 0; i < topics.length; i++)
        {
            int topic = topics[i];
            List<ClientEntry> subscribedClients = this.subscriptions.get(topic);
            if( subscribedClients == null )
            {
                System.out.println("Cannot unsubscribe client"+
                        subscribedClient.name +
                        " - no clients subscribed to " + topic);
            }
            
            if( !subscribedClients.contains(subscribedClient) )
            {
                System.out.println("Cannot unsubscribe client"+
                        subscribedClient.name +
                        " - client not subscribed to " + topic);
            }
            else
            {
                subscribedClients.remove(entry);
                System.out.println("Client "+subscribedClient.name+
                        " unsubscribed to " + topic);
            }
        }
    }
    
    /**
     * Stops the server from listening
     * @throws java.io.IOException
     */
    public void stop()
    {
        this.stopped = true;
        // Stop server - so that we accept no more clients
        if( this.serverSocket != null )
        {
            try
            {
                this.serverSocket.close();
            }
            catch(IOException ioe)
            {
                // ignore
            }
        }
        
        // Stop dispatcher
        this.dispatcher.stop();
        
        // Stop clients
        for (int i = 0; i < clients.size(); i++)
        {
            Sink sink = (Sink)clients.get(i);
            sink.disconnect();
        }
    }
 
    public static void main( String[] args ) throws IOException
    {
        int localPort  = Integer.parseInt(args[0]);
        
        Router router = new Router();
        router.listen(localPort);
    }

}
